package Data_Analysis_llp

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.json4s.scalap.scalasig.ClassFileParser.header

import java.util.Properties

object DataAnalysis1 {
  def main(args: Array[String]): Unit = {
    /*
        DataAnalysis1:分析数据分析岗位分布在全国各地的数量，按照数量进行降序排序
        例如：  上海     158

     */


    //  创建sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("DataAnalysis1")
      .getOrCreate()

    //  准备连接mysql的配置
    val mysql_connect=new Properties()
    mysql_connect.setProperty("user","root")
    mysql_connect.setProperty("password","123456")
    mysql_connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  将清洗完的数据读取到平且创建临时表
  spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/llp?useSSL=false","cleanse",mysql_connect)
    .createOrReplaceTempView("cleanse")

//  数据分析部分:根据位置分组，求和，再根据数量排序(降序)，最后只取出前10
    val result:DataFrame=spark.sql(
      """
        |select
        |position,
        |count(*) as number
        |from cleanse
        |group by  position
        |order by number desc
        |""".stripMargin)

    //  使用dataframe的方法将数据写入到数据库
    result.write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/llp?useSSL=false","DataAnalysis1",mysql_connect)



    //  将分析的结果写入到hdfs(默认是下载到hdfs，后续可以使用hdfs dfs -get将文件下载下来到文件系统)
    //  repartition(1):指定数据文件保存到一个分区
    result.repartition(1).write.mode("overwrite").csv("/llp/DataAnalysis_all/DataAnalysis1.csv")

    //  关闭sparksql的环境
    spark.close()

  }

}
